/*
 * Copyright (C), 2005-2018, 上海瑛麒动漫科技有限公司
 * FileName: RedissonClientImpl.java
 * Author:   tianmu
 * Date:     2018年2月22日 下午1:54:22
 * Description: //模块目的、功能描述
 * History: //修改记录
 * <author>      <time>      <version>    <desc>
 * 修改人姓名             修改时间            版本号                  描述
 */
package com.sfang.common.redis.config;

import com.alibaba.fastjson.JSON;
import com.sfang.common.redis.redisson.*;
import com.sfang.common.redis.sdk.*;
import org.redisson.Redisson;
import org.redisson.api.RBatch;
import org.redisson.api.RBlockingDeque;
import org.redisson.api.RDelayedQueue;
import org.redisson.api.RQueue;
import org.redisson.codec.JsonJacksonCodec;
import org.redisson.config.Config;
import org.redisson.config.SingleServerConfig;
import org.redisson.config.TransportMode;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.ArrayList;
import java.util.List;
import java.util.Set;
import java.util.concurrent.TimeUnit;

/**
 * 〈一句话功能简述〉<br>
 * 〈功能详细描述〉
 *
 * @author tianmu
 * @see [相关类/方法]（可选）
 * @since [产品/模块版本] （可选）
 */
public class RedissonClient implements RedisClient {
    /**
     * Redis的URI头
     */
    private static final String REDIS_PREFIX = "redis://";
    private static String REDIS_KEY_PREFIX = "";
    private static Logger LOGGER = LoggerFactory.getLogger(RedissonClient.class);
    private org.redisson.api.RedissonClient redissonClient;

    public RedissonClient(RedisClusterConfiguration configuration, Integer database, String redisKeyPrefix) {
        try {
            REDIS_KEY_PREFIX = redisKeyPrefix;
            Set<RedisNode> nodes = configuration.getClusterNodes();
            final Config config = new Config();
            config.setTransportMode(TransportMode.NIO);
            config.setCodec(new JsonJacksonCodec());

            if (nodes.size() > 1) {
                config
                        .useClusterServers()
                        .setRetryAttempts(3)
                        .setMasterConnectionPoolSize(500)
                        .setSlaveConnectionPoolSize(500)
                        .setScanInterval(2000)
                        .setIdleConnectionTimeout(10000)
                        .setConnectTimeout(30000)
                        .setTimeout(3000)
                        .addNodeAddress(convertNodes(nodes));


            } else {
                RedisNode redisNode = (RedisNode) configuration.getClusterNodes().toArray()[0];
                SingleServerConfig singleServerConfig = config
                        .useSingleServer()
                        .setAddress(REDIS_PREFIX + redisNode.getHost() + ":" + redisNode.getPort())
                        .setConnectionPoolSize(500)
                        .setPingConnectionInterval(1000)
                        .setRetryAttempts(3)
                        .setIdleConnectionTimeout(10000)
                        .setConnectTimeout(30000)
                        .setTimeout(3000);

                if (database != null)
                    singleServerConfig.setDatabase(database);
            }

            redissonClient = Redisson.create(config);
            LOGGER.info(String.format("redissonClient initalize success"));
        } catch (Exception e) {
            LOGGER.error(String.format("redissonClient initalize failed:%s", JSON.toJSON(e)));
        }
    }

    @Override
    public RBatch createBatch() {
        return redissonClient.createBatch();
    }

    public <T> void addDelayQueue(T value, long delay, String queueCode) {
        addDelayQueue(value, delay, TimeUnit.SECONDS, queueCode);
    }

    public <V> RBlockingDeque<V> getBlockingDeque(String name) {
        return redissonClient.getBlockingDeque(name);
    }

    public <V> RDelayedQueue<V> getDelayedQueue(RQueue<V> destinationQueue) {
        return redissonClient.getDelayedQueue(destinationQueue);
    }

    /**
     * 添加延迟队列
     *
     * @param value     队列值
     * @param delay     延迟时间
     * @param timeUnit  时间单位
     * @param queueCode 队列键
     * @param <T>       泛型
     */
    public <T> void addDelayQueue(T value, long delay, TimeUnit timeUnit, String queueCode) {
        try {
            RBlockingDeque<Object> blockingDeque = redissonClient.getBlockingDeque(queueCode);
            RDelayedQueue<Object> delayedQueue = redissonClient.getDelayedQueue(blockingDeque);
            delayedQueue.offer(value, delay, timeUnit);
            LOGGER.info("(添加延时队列成功) 队列键：{}，队列值：{}，延迟时间：{}", queueCode, value, timeUnit.toSeconds(delay) + "秒");
            //释放队列
            delayedQueue.destroy();
        } catch (Exception e) {
            LOGGER.error("(添加延时队列失败) {}, value {}", e.getMessage(), value);
            throw new RuntimeException("(添加延时队列失败)");
        }
    }

    @Override
    public <V> RedisBucket<V> getBucket(String name) {
        return new RedissonBucket<V>(redissonClient, REDIS_KEY_PREFIX + name);
    }

    @Override
    public <V> RedisList<V> getList(String name) {
        return new RedissonList<V>(redissonClient, REDIS_KEY_PREFIX + name);
    }

    @Override
    public RedisLock getLock(String name) {
        return new RedissonLock(redissonClient, REDIS_KEY_PREFIX + name);
    }

    @Override
    public <K, V> RedisMap<K, V> getMap(String name) {
        return new RedissonMap<K, V>(redissonClient, REDIS_KEY_PREFIX + name);
    }

    @Override
    public RedissonAtomicLong getAtomicLong(String name) {
        return new RedissonAtomicLong(redissonClient, REDIS_KEY_PREFIX + name);
    }

    @Override
    public <V> RedisSet<V> getSet(String name) {
        return new RedissonSet<V>(redissonClient, REDIS_KEY_PREFIX + name);
    }

    @Override
    public void shutdown() {
        if (!redissonClient.isShutdown()) {
            redissonClient.shutdown();
        }
    }

    @Override
    public Iterable<String> keys() {
        Iterable<String> keys = redissonClient.getKeys().getKeys();
        return keys;
    }

    @Override
    public List<String> keys(String pattern) {
        Iterable<String> keys = redissonClient.getKeys().getKeysByPattern(REDIS_KEY_PREFIX + pattern);

        List<String> rlt = new ArrayList<>();
        for (String key : keys) {
            rlt.add(key.replace(REDIS_KEY_PREFIX, ""));
        }

        return rlt;
    }

    @Override
    public void deleteByPattern(String pattern) {
        redissonClient.getKeys().deleteByPattern(REDIS_KEY_PREFIX + pattern);
    }

    @Override
    public RedisScript getScript() {
        return new RedissonScript(redissonClient);
    }

    /**
     * 功能描述: <br>
     * 转换节点成字符串数组
     *
     * @param nodes
     * @return
     * @see [相关类/方法](可选)
     * @since [产品/模块版本](可选)
     */
    private String[] convertNodes(Set<RedisNode> nodes) {
        String[] strNodes = new String[nodes.size()];
        int index = 0;
        for (RedisNode node : nodes) {
            strNodes[index] = REDIS_PREFIX + node.getHost() + ":" + node.getPort();
            index++;
        }
        return strNodes;
    }

}
